package com.zjh.reactive;

import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.GroupedFlux;

import java.util.*;

/**
 * @Auther: zjh
 * @Date: 2025/11/3 - 11 - 03 - 14:16
 * @Description: com.zjh.reactive
 * @version: 1.0
 */
public class BufferDemo {
    @Test
    public void buffer() {
        Flux<List<Integer>> buffer = Flux.range(1, 10000)
                .buffer(100);
        buffer
                .subscribe(System.out::println);
    }

    @Test
    public void window() {
        Flux<Flux<Integer>> window1 = Flux.range(1, 10)                    // 1,2,3,4,5,6,7,8,9,10
                .window(3);// 分成多个Flux，每个最多3个元素
        Flux<List<Integer>> listFlux = window1.flatMap(window -> window.collectList());// 将每个窗口转为List
        listFlux.subscribe(list -> System.out.println("窗口: " + list));


        Flux<Flux<Integer>> fluxFlux = Flux.range(101, 20)
                //当遇到素数开窗口
                .windowUntil(BufferDemo::isPrime, false);
        fluxFlux.subscribe(
                window ->
                        window.collectList()
                                .subscribe(
                                        item -> System.out.println("window:" + item)
                                )
        );

    }

    private static boolean isPrime(Integer integer) {
        double sqrt = Math.sqrt(integer);
        if (integer < 2) {
            return false;
        }
        if (integer == 2 || integer == 3) {
            return true;
        }
        if (integer % 2 == 0) {
            return false;
        }
        for (int i = 3; i <= sqrt; i++) {
            if (integer % i == 0) {
                return false;
            }
        }
        return true;
    }

    @Test
    public void groupBy() throws InterruptedException {
//        Map<String, String> map = new HashMap<>();
//        map.put("奇数","");
//        map.put("偶数","");
//       Flux.range(1, 7)
//                .groupBy(e -> {
//                    System.out.println("处理元素"+e);
//                   return e % 2 == 0 ? "偶数" : "奇数";
//                })
////               .doOnNext(group -> System.out.println("创建分组: " + group.key()))
////               .doOnNext(groupedFlux -> {
////                   String key = groupedFlux.key();
////                   System.out.println(key);
////                   groupedFlux.doOnNext(i->{
////                       System.out.println("处理"+key);
////                       map.put(key,map.get(key)+ i.toString());
////                   }).subscribe(s->{
////                       System.out.print("");
////                   });
////               })
//
////               .flatMap(group -> {
////                   System.out.println("处理分组: " + group.key());
////                   return group.map(i -> "分组" + group.key() + ": " + i);
////               })
//               .subscribe(s->{
//
//
//               });
////        System.out.println(map);
//        Thread.sleep(10000);
        Flux.range(1, 7)
                .groupBy(e -> e % 2 == 0 ? "偶数" : "奇数")
                .flatMap(groupedFlux ->
                        groupedFlux
                                .doOnNext(list -> System.out.println("开始 " + groupedFlux.key() + " 收集完: " + list))
                                .collectList()
                                .doOnNext(list -> System.out.println("组 " + groupedFlux.key() + " 收集完: " + list))
                                .map(list -> Map.entry(groupedFlux.key(), list))
                )
                .doOnNext(entry -> System.out.println("flatMap 发出: " + entry))
                .collectMap(Map.Entry::getKey, Map.Entry::getValue)
                .doOnSuccess(map -> System.out.println("✅ collectMap 收集完成: " + map))
                .subscribe();

    }
    @Test
    public void test(){
        Flux.range(1, 7)
                .groupBy(item -> item % 2 == 0 ? "偶数" : "奇数")
                .subscribe(groupFlux ->
                        groupFlux.scan(
                                        new ArrayList<Integer>(),
                                        (list, element) -> {
                                            list.add(element);
                                            if (list.size() > 2) {
                                                list.remove(0);
                                            }
                                            return list;
                                        }
                                )
                                .filter(list -> !list.isEmpty())
                                .subscribe(list ->
                                        System.out.println(groupFlux.key() + "<====>" + list)
                                )
                );

    }

}


